package com.atguigu.flink.datastream.transform;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import com.atguigu.flink.pojo.Wordcount;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 14:06
 * <p>
 * 聚合算子:
 * 简单聚合  :  sum  min  minBy  max  maxBy
 * 归约聚合  :  reduce
 * 两两聚合 , 同一个key的第一条数据到来，直接输出结果，不进reduce方法， 从第二条数据往后， 每来一条数据都会进reduce方法进行两两聚合。
 * <p>
 * Flink要求:  要聚合，先keyBy
 */
public class Flink03_ReduceTransform {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        ds.print("input");

        //统计每个user的点击次数 ， 通过reduce来实现。
        ds.map(
                event -> new Wordcount(event.getUser(), 1)
        ).keyBy(
                Wordcount::getWord
        ).reduce(
                new ReduceFunction<Wordcount>() {
                    /*
                       两个参数：
                          value1:  要合并的第一个值
                          value2:  要合并的第二个值
                     */
                    @Override
                    public Wordcount reduce(Wordcount value1, Wordcount value2) throws Exception {
                        System.out.println("reduce.....");
                        return new Wordcount(value1.getWord(), value1.getCount() + value2.getCount());
                    }
                }
        );
//        .print("reduce");

        // 记录所有用户中访问频次最高的那个，也就是当前访问量最大的用户是谁。
        // 首先统计每个用户的访问次数
        /**
         * input> Event(user=Zhangs, url=/cart, ts=1686743900641)   第1次来的是 Zhangs
         * max> (word='Zhangs', count=1)  Zhangs是max  count=1
         * input> Event(user=Zhangs, url=/home, ts=1686743901657)  第2次来的是 Zhangs
         * reduce.....
         * value1 = (word='Zhangs', count=1)   value1是第2个reduce 上一次return的，
         * value2 = (word='Zhangs', count=2)   value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=2)   最大的是Zhangs count=2
         * input> Event(user=Jerry, url=pay, ts=1686743902658)  输入的数据是Jerry
         * value1 = (word='Zhangs', count=2)     value1是第2个reduce 上一次return的
         * value2 = (word='Jerry', count=1)      value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=2)    最大的是Zhangs count=2
         * input> Event(user=Zhangs, url=/home, ts=1686743903659)   输入的数据是 Zhangs
         * reduce.....
         * value1 = (word='Zhangs', count=2)    value1是第2个reduce 上一次return的
         * value2 = (word='Zhangs', count=3)    value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=3)     最大的是Zhangs count=3
         * input> Event(user=Lisi, url=detail, ts=1686743904660)   输入的数据是 Lisi
         * value1 = (word='Zhangs', count=3)    value1是第2个reduce 上一次return的
         * value2 = (word='Lisi', count=1)      value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=3)     最大的是Zhangs count=3
         * input> Event(user=Peiqi, url=detail, ts=1686743905660)   输入的数据是 Peiqi
         * value1 = (word='Zhangs', count=3)   value1是第2个reduce 上一次return的
         * value2 = (word='Peiqi', count=1)    value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=3)     最大的是Zhangs count=3
         * input> Event(user=Jerry, url=detail, ts=1686743906662)   输入的数据是 Jerry
         * reduce.....
         * value1 = (word='Zhangs', count=3)     value1是第2个reduce 上一次return的
         * value2 = (word='Jerry', count=2)      value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=3)    最大的是Zhangs count=3
         * input> Event(user=Lisi, url=/home, ts=1686743907663)    输入的数据是 Lisi
         * reduce.....
         * value1 = (word='Zhangs', count=3)     value1是第2个reduce 上一次return的
         * value2 = (word='Lisi', count=2)       value2是第1个reduce算子算出来的
         * max> (word='Zhangs', count=3)     最大的是Zhangs count=3
         */
        ds.map(
                event -> //Wordcount.builder().word(event.getUser()).count(1).build()
                        new Wordcount(event.getUser(), 1)
        ).keyBy(
                Wordcount::getWord
        ).reduce(
                (value1, value2) -> new Wordcount(value1.getWord(), value1.getCount() + value2.getCount())
        ).keyBy(
                wordcount -> true  //将所有的数据keyby到一起。
        ).reduce(
                (value1, value2) -> {

                    System.out.println("value1 = " + value1);
                    System.out.println("value2 = " + value2);

                    if (value1.getCount() > value2.getCount()) {
                        return value1;
                    } else {
                        return value2;
                    }
                }
        )
                .print("max");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
